package com.uptown.matrix.core.consumer.redis;

import java.time.Duration;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.atomic.AtomicReference;

import javax.annotation.Resource;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StreamOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;
import org.springframework.stereotype.Component;

import com.uptown.matrix.core.annotation.MatrixMessageListener;
/**
 * redis stream 配置类,去yml中redis配置
 *
 */
@SuppressWarnings("all")
@Slf4j
@Component
public class RedisConsumerHandler implements DisposableBean {

	@Resource
	StringRedisTemplate stringRedisTemplate;

	@Resource
	RedisConnectionFactory factory;

	@Resource
	Map<String, StreamMessageListenerContainer<String, ObjectRecord<String, Object>>> redisConsumerMap;


	public void createSubscription(StreamListener streamListener, MatrixMessageListener annotation, String groupTopicKey) {

		String streamKey = annotation.streamKey();
		String group = annotation.consumerGroup();
		String consumerName = annotation.consumerName();

		StreamOperations<String, String, Object> streamOperations = this.stringRedisTemplate.opsForStream();

		if (stringRedisTemplate.hasKey(streamKey)) {
			StreamInfo.XInfoGroups groups = streamOperations.groups(streamKey);

			AtomicReference<Boolean> groupHasKey = new AtomicReference<>(false);

			groups.forEach(groupInfo -> {
				if (Objects.equals(group, groupInfo.getRaw().get("name"))) {
					groupHasKey.set(true);
				}
			});

			if (groups.isEmpty() || !groupHasKey.get()) {
				creatGroup(streamKey, group);
			} else {
				groups.stream().forEach(g -> {
					StreamInfo.XInfoConsumers consumers = streamOperations.consumers(streamKey, g.groupName());
				});
			}
		} else {
			creatGroup(streamKey, group);
		}


		StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, Object>> options =
				StreamMessageListenerContainer.
						StreamMessageListenerContainerOptions.builder()
						.batchSize(10)
						.serializer(new StringRedisSerializer())
						.executor(new ForkJoinPool())
						.pollTimeout(Duration.ZERO)
						.targetType(Object.class)
						.build();

		StreamMessageListenerContainer<String, ObjectRecord<String, Object>> listenerContainer = StreamMessageListenerContainer.create(factory, options);
		StreamOffset<String> streamOffset = StreamOffset.create(streamKey, ReadOffset.lastConsumed());

		Consumer consumer = Consumer.from(group, consumerName);

		Subscription subscription = listenerContainer.receive(consumer, streamOffset, streamListener);
		listenerContainer.start();

		this.redisConsumerMap.put(groupTopicKey, listenerContainer);
		return;
	}

	private void creatGroup(String key, String group) {
		StreamOperations<String, String, Object> streamOperations = this.stringRedisTemplate.opsForStream();
		String groupName = streamOperations.createGroup(key, group);
		log.info("redis message queue creat group -> {}", groupName);
	}


	@Override
	public void destroy() {
		this.redisConsumerMap.forEach((streamKey, listener) -> {
			listener.stop();
		});
	}
}
